package com.company.bookstore.test;

import com.alibaba.fastjson.JSON;
import com.company.bookstore.common.redis.RedisUtil;
import com.company.bookstore.entities.order_module.Order;
import com.company.bookstore.qo.order_module.OrderQo;
import com.company.bookstore.service.order_module.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.UUID;

/**
 * The broker[iZbp18sab7k1nntw5cnky0Z, 172.16.32.61:10911] boot success.
 * serializeType=JSON and name server is localhost:9876
 *
 *  消费者组 订阅相同的topic，消费逻辑一致，在消息消费方面实现负载均衡
 *
 *  使用 DefaultMQPushConsumer 主要是设置好各种参数和传人处理消息的函数 。
 *  系统收到消息后自动调用处理函数来处理消息，自动保存 Offset，而且加入新的 DefaultMQPushConsumer后会自动做负载均衡。
 *
 *  RocketMQ支持两种消息模式: Clustering和Broadcasting。
 *
 * 在 Clustering模式下，同一个 ConsumerGroup(GroupName相同) 里的每 个 Consumer
 * 只消费所订阅消 息的一部分 内容， 同一个 ConsumerGroup 里所有的 Consumer消费的内容合起来才是所订阅 Topic 内容的整体，
 * 从而达到负载均衡的目的 。
 * 在 Broadcasting模式下，同一个 ConsumerGroup里的每个 Consumer都
 * 能消费到所订阅 Topic 的全部消息，也就是一个消息会被多次分发，被 多个 Consumer消费。
 *
 * 源码分析
 * https://blog.csdn.net/panxj856856/article/details/80776032
 *
 *  应用介绍
 *  https://blog.csdn.net/weixin_38003389/article/details/86658396
 *
 *  pushMqConsumer  主动拉取
 *
 *  pullMessage函数的参数是 final PullRequest pullRequest ，
 *  这是通过“长轮询”方式达到 Push效果的方法，长轮询方式既有 Pull 的优点，又兼具 Push方式的实时性。
 *
 *  “长轮询”的 主动权还是 掌 握在 Consumer 手中， Broker 即使有大 量 消息积 压 ，也不会主动推 送给 Consumer 。
 *
 *  上面我们分析得知PushConsumer的核心还是 Pull 方式。
 *
 *  每个 Message Queue 都会有个对应的 ProcessQueue 对象，保存了这个 Message Queue 消息处理状态的快照 。
 *
 *
 *  ProcessQueue对象里主要的内容是一个 TreeMap 和一个读写锁。 TreeMap 里以 Message Queue 的 Offset作为 Key，
 *  以消息内容的引用为 Value，保存了 所有从 MessageQueue 获取到，但是还未被处理的消息;
 *  读写 锁控制着多个线程对 TreeMap 对象的并发访 问
 *
 *  Pull
 *
 *  获取 Message Queue 并遍历 ，从一个 Message Queue 里拉取消息的时候，要传人 Offset参数( long类型 的值)，
 *  随着不断读取消息 ， Offset会不断增长 。 这个时候由用户负责把 Offset 存储下来，根据具体情况可以存到内存里、写到磁盘或者数据库里等 。
 *
 *
 *  拉取消息的请求发出后，会返回: FOUND、 NO_MATCHED_MSG、
 *  NO_NEW_MSG、 OFFSET_ILLEGAL 四种状态，需要根据每个状态做不同的处理 。
 *  比较重要的两个状态是 FOUND 和 NO NEW MSG ，分别表示获取到消息和没 有新的消息 。
 * */

@Component
@Slf4j
public class CustomConsumer implements InitializingBean {
    private  DefaultMQPushConsumer consumer;
    @Resource
    OrderService orderService;
    @Resource
    RedisUtil redisUtil;
    @Override
    public void afterPropertiesSet() throws Exception {
        consumer  = new DefaultMQPushConsumer("group-test");
        consumer.setNamesrvAddr("101.37.33.148:9876");
        consumer.subscribe("topic_second_kill","book");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                msgs.forEach(
                        v->{
                            Order o= JSON.parseObject(v.getBody(), Order.class);
                            o.setOrderNum(UUID.randomUUID().toString());
                            o.setCreateTime(new Date(System.currentTimeMillis()));
                            o.setDeleted(0);

                            orderService.save(o);


//                            log.info(v.toString());
                        }
                );
                log.info("{} Receive New Messages: {}", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
        consumer.start();
    }
//    private static  final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group-test");
//    static {
//        consumer.setNamesrvAddr("192.168.58.128:9876");
//    }
//    public  void receive(String method) {
//        // Instantiate with specified consumer group name.
//
//        try {
//            // Subscribe one more more topics to consume.
//            consumer.subscribe("topic-test",method);
//            // Register callback to execute on arrival of messages fetched from brokers.
//            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
//                log.info("{} Receive New Messages: {}", Thread.currentThread().getName(), msgs);
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            });
//        } catch (MQClientException e) {
//            e.printStackTrace();
//            consumer.shutdown();
//        }
//    }
}
